package studio.raptor.dispatcher.server.task;

import java.util.concurrent.locks.LockSupport;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.dispatcher.server.context.HandlerContext;
import studio.raptor.dispatcher.server.dto.Consumer;
import studio.raptor.dispatcher.server.handler.dispatch.FetchEventHandler;
import studio.raptor.dispatcher.server.handler.dispatch.SendHanlder;

/**
 * @author jack
 * @since 1.0
 */
@Component
@Scope("prototype")
public class DispatcherTask implements Runnable {

  private static final Logger log = LoggerFactory.getLogger(DispatcherTask.class);

  private Consumer consumer;

  private volatile boolean start;

  @Autowired
  FetchEventHandler fetchEventHandler;

  @Autowired
  SendHanlder dispatchHanlder;

  @PostConstruct
  public void init() {
    fetchEventHandler.setSuccessor(dispatchHanlder);
  }

  @PreDestroy
  public void destory() {
    log.info("the consumer:{} dispatcher task closed", consumer.toString());
    setStart(false);
  }


  @Override
  public void run() {
    log.info("the consumer:{},kindId:{},state:{}", consumer.getCode(), consumer.getKindId(),
        start ? "start" : "stop");
    try {
      HandlerContext context = buildContext();
      while (!Thread.currentThread().isInterrupted()) {
        if (isStart()) {
          fetchEventHandler.process(context);
        } else {
          //任务暂停，休眠5s,防止cpu过高
          LockSupport.parkUntil(System.currentTimeMillis() + 500);

        }
      }

    } catch (Exception e) {
      log.error("the consumer->{}-{},consume->{} error", consumer.getCode(), consumer.getName(),
          consumer.getKindId(), e);
    }
  }


  public Consumer getConsumer() {
    return consumer;
  }

  public void setConsumer(Consumer consumer) {
    this.consumer = consumer;
  }

  public boolean isStart() {
    return start;
  }

  public synchronized void setStart(boolean start) {
    this.start = start;
  }

  private HandlerContext buildContext() {
    HandlerContext context = new HandlerContext();
    context.setCode(consumer.getCode());
    context.setName(consumer.getName());
    context.setKindId(consumer.getKindId());
    context.setEventType(consumer.getEventType());
    context.setSort(consumer.isSort());
    return context;
  }
}
